package analysis;

import beans.AdClickEvent;
import beans.AdCountViewByProvince;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.net.URL;
import java.sql.Timestamp;

/**
 * @author zkq
 * @date 2022/10/6 18:36
 */
//页面广告点击量统计
public class AdCountByProvince {
    public static void main(String[] args) throws Exception {
        String path = AdCountByProvince.class.getResource("/AdClickLog.csv").getPath();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<AdClickEvent> inputStream = env.readTextFile(path)
                .map(data -> {
                    String[] fields = data.split(",");
                    return new AdClickEvent(new Long(fields[0]), Long.valueOf(fields[1]), fields[2], fields[3], new Long(fields[4]));
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<AdClickEvent>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<AdClickEvent>() {
                            @Override
                            public long extractTimestamp(AdClickEvent element, long recordTimestamp) {
                                return element.getTimestamp()*1000;
                            }
                        })
                );
        //根据province分组开窗聚合
        SingleOutputStreamOperator<AdCountViewByProvince> aggregate = inputStream
                .keyBy(data -> data.getProvince())
                .window(SlidingEventTimeWindows.of(Time.hours(1), Time.seconds(60)))
                .aggregate(new AdCountAgg(), new AdCountResult());

        aggregate.print();
        env.execute("ad count job");
    }
    public static class AdCountAgg implements AggregateFunction<AdClickEvent,Long,Long>{

        @Override
        public Long createAccumulator() {
            return 0L;
        }

        @Override
        public Long add(AdClickEvent value, Long accumulator) {
            return accumulator + 1;
        }

        @Override
        public Long getResult(Long accumulator) {
            return accumulator;
        }

        @Override
        public Long merge(Long a, Long b) {
            return a + b;
        }
    }
    public static class AdCountResult extends ProcessWindowFunction<Long,AdCountViewByProvince,String, TimeWindow>{

        @Override
        public void process(String s, Context context, Iterable<Long> elements, Collector<AdCountViewByProvince> out) throws Exception {
            String windowEnd = new Timestamp(context.window().getEnd()).toString();
            Long count = elements.iterator().next();
            String province = s;
            out.collect(new AdCountViewByProvince(province,windowEnd,count));
            Thread.sleep(2000);
        }
    }
}
